package com.kai.mqtt.server;

import com.kai.mqtt.config.BrokerProperties;
import com.kai.mqtt.handler.BrokerHandler;
import com.kai.mqtt.handler.MsgCountHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.timeout.IdleStateHandler;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;

/**
 * @author weijk
 * @since 2024/6/2
 */
@Slf4j
@Component
@EnableConfigurationProperties(BrokerProperties.class)
public class BrokerServer {

    @Autowired
    private ApplicationContext applicationContext;

    @Autowired
    private BrokerProperties brokerProperties;

    private EventLoopGroup bossGroup;

    private EventLoopGroup workerGroup;

    @PostConstruct
    public void start() {
        bossGroup = new NioEventLoopGroup();
        workerGroup = new NioEventLoopGroup();

        mqttServer();
    }

    private void mqttServer() {
        try {
            log.info("Initializing MQTT Broker ..., port: {}", brokerProperties.getPort());
            new ServerBootstrap()
                    .group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<>() {
                        @Override
                        protected void initChannel(Channel ch) throws Exception {
                            ch.pipeline()
                                    .addLast("idle", new IdleStateHandler(0, 0, brokerProperties.getKeepAlive()))
                                    .addLast("decoder", new MqttDecoder(brokerProperties.getMaxBytesInMessage()))
                                    .addLast("encoder", MqttEncoder.INSTANCE)
                                    .addLast("msgCount", applicationContext.getBean(MsgCountHandler.class))
                                    .addLast("broker", applicationContext.getBean(BrokerHandler.class));
                        }
                    }).option(ChannelOption.SO_BACKLOG, brokerProperties.getSoBacklog())
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .bind(brokerProperties.getPort()).sync().channel();
            log.info("MQTT Broker Initializing Success, port: {}", brokerProperties.getPort());
        } catch (Exception e) {
            log.error("MQTT Broker Initializing Error, port: {}, message:{}", brokerProperties.getPort(), e.getMessage(), e);
            throw new RuntimeException(e);
        }
    }
}
